// -*-mode:c++; coding:utf-8-*-

#ifndef _XBASE_SSTABLE_HPP_
#define _XBASE_SSTABLE_HPP_

#include <boost/pool/detail/singleton.hpp>
#include <boost/thread/mutex.hpp>

#include <logger.hpp>
#include <uuid.hpp>
#include <wrapper.hpp>
#include <compressor.hpp>
#include <fs.hpp>
#include <kv.hpp>
#include <binary_iarchive.hpp>
#include <binary_oarchive.hpp>

#include "sstable_block_cache.hpp"
#include "kv_scanner.hpp"

namespace xbase
{

	struct sstable_trailer
	{
		enum
		{
			ST_MAGIC = 0x3ee2c230U,
			ST_LENGTH = sizeof(uint32_t) + sizeof(uint16_t) * 2 + sizeof(uint64_t) * 9 // 10 * uint64_t = 80 bytes
		};

		uint32_t m_magic;
		uint16_t m_version;
		uint16_t m_compression_type;
		uint64_t m_data_size; // all data size
		uint64_t m_data_zsize; // all zdata size
		uint64_t m_file_size;  // file size
		uint64_t m_block_size; // user specified value
		uint64_t m_data_block_number;
		uint64_t m_data_record_number;
		uint64_t m_index_key_offset;
		uint64_t m_index_offset_offset;
		uint64_t m_timestamp;

		sstable_trailer(uint16_t version = 0,
				uint16_t compression_type = compressor::NONE,
				uint64_t blocksize = 0ULL)
			: m_magic(ST_MAGIC)
			, m_version(version)
			, m_compression_type(compression_type)
			, m_data_size(0)
			, m_data_zsize(0)
			, m_file_size(0)
			, m_block_size(blocksize)
			, m_data_block_number(0)
			, m_data_record_number(0)
			, m_index_key_offset(0)
			, m_index_offset_offset(0)
			, m_timestamp(0) {}

		inline
		bool valid() const {
			return (m_magic == ST_MAGIC) &&
				(m_index_offset_offset > m_index_key_offset) &&
				(m_file_size - ST_LENGTH > m_index_offset_offset);
		}

		template<typename IArchive>
		inline
		bool load(IArchive &in);
		
		template<typename OArchive>
		inline
		bool save(OArchive &out) const;
		
		std::string to_string() const;
		std::string stat_info() const;
	};

	struct sstable_block_header
	{
		enum block_type	// and magic
		{
			BT_DATA_BLOCK = 0x7203fecdU,
			BT_INDEX_KEY_BLOCK = 0xbc1e52dfU,
			BT_INDEX_OFFSET_BLOCK = 0xab8316a2U
		};

		enum
		{
			SBH_LENGTH = sizeof(uint32_t) + sizeof(uint16_t) + sizeof(uint64_t) * 2
		};

		uint32_t m_magic;
		uint16_t m_compression_type;
		uint64_t m_data_size;
		uint64_t m_data_zsize;

		sstable_block_header() {}
		
		sstable_block_header(block_type type)
			: m_magic(type) {}

		block_type type() const {
			return static_cast<block_type>(m_magic);
		}

		bool valid(block_type type) const {
			return (type == static_cast<block_type>(m_magic));
		}

		template<typename IArchive>
		inline
		bool load(IArchive &in) {
			in >> m_magic
			   >> m_compression_type
			   >> m_data_size
			   >> m_data_zsize;
			return in.good();
		}
		
		template<typename OArchive>
		inline
		bool save(OArchive &out) const {
			out << m_magic
			    << m_compression_type
			    << m_data_size
			    << m_data_zsize;
			return out.good();
		}

		std::string to_string() const;
		std::string stat_info() const;
	};

	struct sstable_index_entry
	{
		sstable_index_entry(const key_wrapper *key,
				    const fs::offset_t &offset)
			: m_key(key)
			, m_offset(offset) {}
		
		const key_wrapper *m_key;
		fs::offset_t m_offset;
	};

	struct sstable_index_entry_less_than
		: public key_less_than
	{
		bool operator()(const sstable_index_entry &entry1,
				const sstable_index_entry &entry2) const {
			return this->key_less_than::operator()(entry1.m_key, entry2.m_key);
		}
		
		bool operator()(const sstable_index_entry &entry,
				const key_wrapper *key) const {
			return this->key_less_than::operator()(entry.m_key, key);
		}
		
		bool operator()(const key_wrapper *key,
				const sstable_index_entry &entry) const {
			return this->key_less_than::operator()(key, entry.m_key);
		}
	};

	enum sstable_type
	{
		ST_NORMAL_SSTABLE, // 正常
		ST_REF_TOP_SSTABLE, // 引用上半部分
		ST_REF_BOTTOM_SSTABLE // 引用下半部分
	};

	static inline
	const char *sstable_type_name(sstable_type type) {
		const char * name = NULL;
		switch(type)
		{
		case ST_NORMAL_SSTABLE:
			name = "ST_NORMAL_SSTABLE";
			break;
		case ST_REF_TOP_SSTABLE:
			name = "ST_REF_TOP_SSTABLE";
			break;
		case ST_REF_BOTTOM_SSTABLE:
			name = "ST_REF_BOTTOM_SSTABLE";
			break;
		default:
			name = "UNKNOWN";
			break;
		}
		return name;
	}

	// 判断子表是哪部分，根据TabletId的奇偶性：偶数是上半部分，奇数
	// 为下半部分
	static inline
	sstable_type ref_to_which(uint64_t tablet_id) {
		const uint64_t mask = 0x01ULL;
		const bool top_half = ((tablet_id & mask) == 0);
		return (top_half
			? ST_REF_TOP_SSTABLE
			: ST_REF_BOTTOM_SSTABLE);
	}

	class sstable
		: private boost::noncopyable
	{
	public:
		struct tag {};
		typedef boost::details::pool::singleton_default< xbase::uuid<tag> > uuid;
		
		typedef boost::intrusive_ptr<sstable> pointer;
		typedef boost::fast_pool_allocator<sstable> self_allocator_type;
		struct sstable_factory
		{
			typedef sstable::self_allocator_type allocator_type;

			static
			allocator_type s_allocator;

			static inline
			sstable *construct(const std::string &real_path,
					   const std::string &sstable_path,
					   const std::string &compacting_sstable_path,
					   const std::string &deleted_sstable_path,
					   const std::string &start_row,
					   const std::string &end_row,
					   uint64_t sstable_sequence_number,
					   uint64_t block_size,
					   compressor::compressor_code compressor,
					   sstable_type type) {
				sstable *p = s_allocator.allocate();
				new(p) sstable(real_path,
					       sstable_path,
					       compacting_sstable_path,
					       deleted_sstable_path,
					       start_row,
					       end_row,
					       sstable_sequence_number,
					       block_size,
					       compressor,
					       type);
				return p;
			}

			static inline
			void destroy(sstable *p) {
				p->~sstable();
				s_allocator.deallocate(p);
			}
		};
		typedef sstable_factory factory_type;

		typedef sstable_index_entry index_entry;
		typedef sstable_index_entry_less_than index_entry_less_than;
		typedef std::vector<index_entry> index_type;

	protected:
		// 
		// configure section
		//
		struct config
		{
			static const uint32_t sstable_version = 0;

			// 性能最好的读/写块大小，用来优化SSTable文件的读写
			static const uint64_t read_block_size = 2ULL * 1024ULL * 1024ULL; // 2M
			static const uint64_t write_block_size = 4ULL * 1024ULL * 1024ULL; // 4M
		};

	protected:
		typedef kb::binary_iarchive<kb::network_byteorder> iarchive_type;
		typedef kb::binary_oarchive<kb::network_byteorder> oarchive_type;

		static inline
		iarchive_type get_iarchive(const boost::asio::const_buffer &buffer) {
			return iarchive_type(boost::asio::buffer_cast<const char*>(buffer),
					     boost::asio::buffer_size(buffer));
		}

		static inline
		oarchive_type get_oarchive(const boost::asio::mutable_buffer &buffer) {
			return oarchive_type(boost::asio::buffer_cast<char*>(buffer),
					     boost::asio::buffer_size(buffer));
		}

	public:
		// 加载已有的sstable文件
		//
		// 状态置为读
		//
		// 注意：加载过程没有锁保护
		static
		pointer load(const std::string &real_path,
			     const std::string &sstable_path,
			     const std::string &deleted_sstable_path,
			     const std::string &start_row,
			     const std::string &end_row,
			     uint64_t sstable_sequence_number,
			     sstable_type type);

		// 
		// 注意：
		// 创建/写sstable过程中，没有锁保护。即
		// create() -> append() -> append_eof()
		// 

		// 用于创建新的sstable文件
		//
		// 状态置为写
		//
		// 要求能通过以下调用序列创建一个新的sstable：
		// sstable::create()
		// foreach key/value
		//     append into sstable
		// append_eof()
		static
		pointer create(const std::string &sstable_path,
			       const std::string &compacting_sstable_path,
			       const std::string &deleted_sstable_path,
			       const std::string &start_row,
			       const std::string &end_row,
			       uint64_t sstable_sequence_number,
			       uint64_t block_size,
			       compressor::compressor_code compressor);

		// 向文件写入记录
		inline
		bool append(const block_entry &entry) {
			return append(entry.m_key, entry.m_value);
		}

		bool append(const key_wrapper *, const value_wrapper *);


		// 写入完毕，将数据写入到文件中，形成完整的sstable文件；但索引
		// 不清空，这样不用再次加载即可使用。
		//
		// 状态置为读
		bool append_eof ();


		// interface

		// 有锁保护
		bool set_deleted();

		//
		// 读写接口
		//

		// 返回sstable扫描器
		kv_scanner::pointer get_scanner(const kv_scan_context::pointer &context);

		std::string to_string () const;

		uint64_t sequence_number() const {
			return m_sequence_number;
		}

		std::string path() const {
			return m_sstable_path;
		}

		// 返回文件中的数据大小，返回的是大约值，用于判断是否分裂及取中间行。
		uint64_t file_size() const {
			return m_trailer.m_file_size;
		}

		std::string get_split_row () const {
			return m_split_row;
		}

		bool is_ref_sstable () const {
			return m_sstable_type != ST_NORMAL_SSTABLE;
		}

		bool is_ref_top_sstable() const {
			return m_sstable_type == ST_REF_TOP_SSTABLE;
		}

		bool is_ref_bottom_sstable() const {
			return m_sstable_type == ST_REF_BOTTOM_SSTABLE;
		}

		~sstable();

		bool get_entry(sstable_block_cache &block,
			       block_entry &entry,
			       std::pair<std::size_t, std::size_t> &index_pair);

		// 将索引指向下一条记录位置，使得通过get_entry可以读到下一条记录
		//
		// 实现为只简单的增加块内索引，块是否读完等检查工作，在get_entry中做
		//
		// 专门为配合get_entry而实现
		void advance_index(std::pair<std::size_t, std::size_t> &index_pair) const {
			index_pair.second += 1;
		}

		// 查找指定的Key属于哪个块中的哪个位置，并顺便返回那个块
		bool lower_bound(sstable_block_cache &got_block,
				 std::pair<std::size_t, std::size_t> &index_pair,
				 const key_wrapper *key);

	protected:
		sstable(const std::string &real_path,
			const std::string &sstable_path,
			const std::string &compacting_sstable_path,
			const std::string &deleted_sstable_path,
			const std::string &start_row,
			const std::string &end_row,
			uint64_t sstable_sequence_number,
			uint64_t block_size,
			compressor::compressor_code compressor,
			sstable_type type)
			: m_id(uuid::instance().get())
			, m_sstable_path(sstable_path)
			, m_compacting_sstable_path(compacting_sstable_path)
			, m_deleted_sstable_path(deleted_sstable_path)
			, m_start_row(start_row)
			, m_end_row(end_row)
			, m_sequence_number(sstable_sequence_number)
			, m_sstable_type(type)
			, m_deleted(false)
			, m_real_path(real_path)
			, m_file(fs::BAD_FILE)
			, m_file_buffer()
			, m_trailer(config::sstable_version,
				    static_cast<uint16_t>(compressor),
				    block_size)
			, m_split_row()
			, m_index_begin(0)
			, m_index_end(0)
			, m_write_status(0) {}
		
		bool check_file_for_read() const;
		bool check_file_for_write() const;

		bool load_data(const boost::asio::mutable_buffer &buffer,
			       fs::offset_t offset) const;

		inline
		bool append_data_block(const boost::asio::const_buffer &data) {
			return append_block(data,
					    sstable_block_header::BT_DATA_BLOCK);
		}

		inline
		bool append_last_data_block(const boost::asio::const_buffer &data) {
			return append_block(data,
					    sstable_block_header::BT_DATA_BLOCK);
		}

		bool append_index_key_block(const boost::asio::const_buffer &data) {
			// 记录索引Key的起始偏移
			m_trailer.m_index_key_offset = m_trailer.m_file_size;
			return append_block(data,
					    sstable_block_header::BT_INDEX_KEY_BLOCK);
		}

		bool append_index_offset_block(const boost::asio::const_buffer &data) {
			// 记录索引Offset的起始偏移
			m_trailer.m_index_offset_offset = m_trailer.m_file_size;
			return append_block(data,
					    sstable_block_header::BT_INDEX_OFFSET_BLOCK);
		}
		
		// while data comes big enough, it'll be flush into file
		bool append_block(const boost::asio::const_buffer &data,
				  sstable_block_header::block_type block_type);

		bool append_trailer();

		bool update_split_row_and_filesize();

		// for sstable::load()
		void init_for_loading() {}

		// for create
		void init_for_creating() {
			// init for write
			m_write_status = &__inner_write_status;
			m_write_status->m_uncompressed_block_size = m_trailer.m_block_size;
			m_write_status->m_last_key = NULL;
			// 为索引预分配一个块大小的内存
			m_write_status->m_tmp_index_key_buffer.prepare(m_trailer.m_block_size);
			m_write_status->m_tmp_index_offset_buffer.prepare(m_trailer.m_block_size);
			// 预分配空间2倍块大小，避免生成块时导致内存重分配和拷贝
			m_write_status->m_tmp_data_block_buffer.prepare(m_trailer.m_block_size << 1);

			// 如果块大小比缓存大小大，则该缓存将不启用，以避免不必要的内存拷贝
			if(m_trailer.m_block_size < config::write_block_size)
			{
				m_write_status->m_tmp_data_buffer.prepare(config::write_block_size +
									  m_trailer.m_block_size);
			}
		}

		bool load_trailer();
		bool load_index();
		bool build_index(boost::asio::const_buffer key,
				 boost::asio::const_buffer offset);
		void add_index(const key_wrapper *key, const fs::offset_t &offset);

		sstable_block_cache get_block(const index_type::const_iterator &index) const {
			return get_block(index - m_index.begin());
		}

		// 数据操作过程中，保有该函数中有锁保护
		sstable_block_cache get_block(std::size_t index) const;

		bool check_index(std::size_t index) const {
			return (index >= m_index_begin) && (index < m_index_end);
		}

	private:
		const int m_id;	// 系统唯一的SSTableID,用于区别不同的SSTable，提高缓存管理的效率
		
		const std::string m_sstable_path; // self file
		const std::string m_compacting_sstable_path; // compacting-flag tagged in sstable path
		const std::string m_deleted_sstable_path; // deleted-flag tagged in sstable path
		const std::string m_start_row;
		const std::string m_end_row;
		const uint64_t m_sequence_number; // SSTable的序列号，从文件名中取得

		const sstable_type m_sstable_type;

		bool m_deleted;

		// if ref-sstable, it's ref'ed sstable path; or
		// compacting sstable path if compacting; or deleted
		// sstable path if deleted; otherwise equal to sstable
		// path
		std::string m_real_path;
		mutable fs::file_t m_file; // 文件句柄
		mutable simple_buffer m_file_buffer; // 读写时，作为压缩解压数据的缓存区
		mutable boost::mutex m_file_lock;

		// 下面三个域在build_index完成的时候设置:update_split_row_and_filesize
		sstable_trailer m_trailer; // 如果是引用文件，其中的成员变量应该被更新
		std::string m_split_row;
		// 如果是引用文件，分界线上的块可能被两个子文件共享
		// 索引范围
		// [m_index_begin, m_index_end)
		std::size_t m_index_begin;
		std::size_t m_index_end;

		simple_buffer m_index_key_data; // 持有Index Key的数据，在创建SSTable时会同时用作写Key的缓存
		index_type m_index;

		// write status
		struct write_status
		{
			simple_buffer m_tmp_index_key_buffer;
			simple_buffer m_tmp_index_offset_buffer; // 只在创建表时会用作临时存储
			simple_buffer m_tmp_data_block_buffer;
			simple_buffer m_tmp_data_buffer;    // 写文件缓存，数据会积累到这里，达到一定大小才真正写入文件
			uint64_t m_uncompressed_block_size; // 估计值
			const key_wrapper *m_last_key;

			void clear() {
				m_tmp_index_key_buffer.release();
				m_tmp_index_offset_buffer.release();
				m_tmp_data_block_buffer.release();
				m_tmp_data_buffer.release();
				m_uncompressed_block_size = 0;
				m_last_key = NULL;
			}
		};
		write_status __inner_write_status; // 改为用成员变量，就不需要分配这个小对象了
		write_status *m_write_status; // == 0, not writable; or == &__inner_write_status, writable

		reference_counter m_reference_counter;
		INTRUSIVE_PTR_HOOKS(sstable, m_reference_counter, factory_type);

		// ...
	};


} // namespace xbase

#include "sstable.ipp"

#endif	// _XBASE_SSTABLE_HPP_
